/*
 * ebcd_snr.c - EBC - callback handlers and functions for ebcd 
 * Clunix, KyungWoon Cho, cezanne@clunix.com, 2006.1
 *
 * Copyright (c) 2001-2006 by Clunix Inc. All Rights Reserved.
 */

#include "config.h"

#include "ebcd.h"

/* TODO:: */
#define PKT_MAX_LEN	1472

ebc_pktb_t *
ebcd_build_pktb(ebc_addr_t *paddr_src, ebc_addr_t *paddr_dst,
		sr_t *sr_data, ebc_msgt_t msgt)
{
	ebc_pktb_t	*pktb;
	ebc_mhdr_t	*mhdr;

	pktb = ebc_pktb_new();
	mhdr = &pktb->pkthdr.mhdr;
	if (paddr_src)
		mhdr->srcaddr = *paddr_src;
	else
		mhdr->srcaddr = ebcd_myaddr;

	mhdr->dstaddr = *paddr_dst;
	mhdr->msgid = ebc_get_msgid(ebcd_mch);
	mhdr->msgt = msgt;
	pktb->pkthdr.clid = ebcd_clid;
	pktb->pkthdr.pktlen = EBC_PKTHDR_LEN;

	if (sr_data) {
		char		*buf;
		ecm_size_t	buflen;
		sr_enc_ctx_t	enc_ctx;

		buf = pktb->pktd->data + EBC_PKTHDR_LEN;
		buflen = PKT_MAX_LEN - EBC_PKTHDR_LEN;
		
		enc_ctx = sr_encode_init(sr_data, NULL, SR_RECUR);
		if (sr_encode_ex(buf, &buflen, enc_ctx))
			pktb->pkthdr.pktlen += buflen;
		else {
			ECMASSERT(0);
		}
		sr_encode_done(enc_ctx);
	}

	ebc_pkthdr_encode(pktb);

	return pktb;
}

ebc_err_t
ebcd_push_pktb(ebcd_pch_t *dpch, ebc_pktb_t *pktb)
{
	ebc_pch_t	*pchp;
	ebc_err_t	err;

	pchp = dpch->pchp;
	ecm_mutex_lock(pchp->mutex);
	if (pktb)
		pch_push_pktb(pchp, pktb);
	ecm_mutex_unlock(pchp->mutex);
	err = pch_write_all(pchp);
	switch (err) {
	case EBC_OK:
		set_MRU_sent_dpch(dpch);
		break;
	case -ERR_EBC_AGAIN:
		err = EBC_OK;
		break;
	default:
		break;
	}
	return err;
}

ebc_err_t
ebcd_push_msg(ebcd_pch_t *dpch, ebc_msg_t *msg)
{
	ebc_pch_t	*pchp = dpch->pchp;
	ebc_err_t	err;

	err = pch_push_msg(pchp, msg);
	switch (err) {
	case EBC_OK:
		set_MRU_sent_dpch(dpch);
		break;
	case -ERR_EBC_AGAIN:
		break;
	default:
		break;
	}
	return err;
}

ebc_err_t
ebcd_send_msg(ebcd_pch_t *dpch, ebc_msg_t *msg)
{
	ebc_err_t	err;

	err = pch_send_msg(dpch->pchp, msg);
	if (err == EBC_OK)
		set_MRU_sent_dpch(dpch);
	return err;
}

static void
push_msg(ebc_addr_t *saddr, ebc_addr_t *daddr,
	 ebc_msgid_t msgid, ebc_msgt_t msgt)
{
	ebcd_pch_t	*dpch;
	ebc_msg_t	*msg;

	dpch = get_routed_pch(daddr);
	if (dpch == NULL)
		return;

	if (saddr == NULL)
		saddr = &ebcd_myaddr;
	msg = ebc_msg_build(saddr, daddr, msgt, msgid, NULL);
	ebcd_push_msg(dpch, msg);
	ebcd_pch_put(dpch);
	ebc_msg_free(msg);
}

ebc_err_t
ebcd_send_aliveness(ebcd_pch_t *dpch)
{
	set_MRU_sent_dpch(dpch);
	return pch_trysend_pktb(dpch->pchp, dpch->pktb_aliveness);
}

void
ebcd_push_errmsg(ebc_addr_t *saddr, ebc_addr_t *daddr,
		 ebc_msgid_t msgid, ecm_uint32 err_msgt, ECMBOOL req)
{
	ebc_msgt_t	msgt;

	msgt = EBC_MSGT_ONEWAY | EBC_MSGT_ERR2MSGT(err_msgt);
	if (req)
		msgt |= EBC_MSGT_REQ;
	push_msg(saddr, daddr, msgid, msgt);
}

ECMBOOL
ebcd_push_bdcastack(ebcd_pch_t *dpch, ebc_mhdr_t *mhdr)
{
	ebc_pktb_t	*pktb;
	ebc_mhdr_t	*mhdr_new;
	ecm_uint32	donecnt = n_ext_pchs;

	if (mhdr->srcaddr.mdid != mhdr->dstaddr.mdid)
		donecnt++;
	pktb = ebc_pktb_new();
	mhdr_new = &pktb->pkthdr.mhdr;
	mhdr_new->srcaddr = ebcd_myaddr;
	mhdr_new->dstaddr = mhdr->srcaddr;
	mhdr_new->msgid = mhdr->msgid;
	mhdr_new->msgt = EBC_MSGT_ONEWAY | EBC_MSGT_BDCASTACK;
	pktb->pkthdr.clid = dpch->pchp->clid;
	pktb->pkthdr.pktlen = EBC_PKTHDR_LEN + 4;
	ebc_pkthdr_encode(pktb);

	*(ecm_uint32 *)(pktb->pktd->data + EBC_PKTHDR_LEN) = ecm_htonl(donecnt);

	EBCDDBG(DBG_BDCAST, "bdcast ack: %s donecnt: %u", ebc_dbg_pktb(pktb), donecnt);
	ebcd_push_pktb(dpch, pktb);
	if (donecnt == 0)
		return ECM_FALSE;
	return ECM_TRUE;
}

void
ebcd_cancel_msg(ebcd_pch_t *dpch)
{
	ecm_ht_iterator_t	iterator;
	ebc_msg_t	*msg;

	ecm_mutex_lock(ebcd_mch->outmsgs_lock);
	for (msg = ecm_hashtable_first(ebcd_mch->outmsgs_hash, &iterator, NULL);
	     msg;
	     msg = ecm_hashtable_next(ebcd_mch->outmsgs_hash, &iterator, NULL)) {
		if (msg->mhdr.dstaddr.ndid == dpch->pchp->ndid) {
			msg->cancelled = ECM_TRUE;
			ecm_cond_signal(msg->cond);
		}
	}
	PCH_CANCEL(dpch->pchp);
	ecm_mutex_unlock(ebcd_mch->outmsgs_lock);
}

static ebc_err_t
ebcd_recv_msg(ebc_msg_t *msg, ecm_uint32 timeout)
{
	while (!msg->cancelled && msg->donecur < msg->donecnt) {
		if (!ecm_cond_wait(msg->cond, ebcd_mch->outmsgs_lock, &timeout))
			return -ERR_EBC_TIMEOUT;
	}

	if (msg->cancelled) 
		return -ERR_EBC_CANCELLED;
	if (msg->donecnt == 0)
		return -ERR_EBC_NOTEXIST;
	switch (EBC_MSGT_ERR(msg->mhdr.msgt)) {
	case 0:
		return EBC_OK;
	case EBC_MSGT_ERR_ROUTE:
	default:
		return -ERR_EBC_NOTEXIST;
	}
}

static ebc_err_t
ebcd_snr_msg(ebcd_pch_t *dpch, ebc_msg_t *msg, ecm_uint32 timeout)
{
	ebc_err_t	err;

	ecm_mutex_lock(dpch->mutex);
	if (dpch->destroyed) {
		ecm_mutex_unlock(dpch->mutex);
		return -ERR_EBC_DISCONN;
	}
	ecm_mutex_lock(ebcd_mch->outmsgs_lock);
	ebc_reg_outmsg(ebcd_mch, msg);
	ecm_mutex_unlock(dpch->mutex);

	if ((err = ebcd_send_msg(dpch, msg)) == EBC_OK)
		err = ebcd_recv_msg(msg, timeout);

	ebc_unreg_outmsg(ebcd_mch, msg);
	ecm_mutex_unlock(ebcd_mch->outmsgs_lock);
	return err;
}

ebc_err_t
ebcd_call(ebc_addr_t *daddr, sr_t **psr, ebc_msgt_t msgt, ecm_uint32 timeout)
{
	ebcd_pch_t	*dpch;
	ebc_msg_t	*msg;
	ebc_msgid_t	msgid;
	ebc_err_t	err;

	msgid = ebc_get_msgid(ebcd_mch);
	msg = ebc_msg_build(&ebcd_myaddr, daddr, EBC_MSGT_REQ | msgt, msgid, *psr);
	dpch = get_routed_pch(daddr);
	if (dpch == NULL)
		return -ERR_EBC_NOROUTE;
	if (msgt & EBC_MSGT_ONEWAY)
		err = ebcd_send_msg(dpch, msg);
	else {
		err = ebcd_snr_msg(dpch, msg, timeout);
		if (err == EBC_OK) {
			sr_free(*psr);
			*psr = msg->sr_data;
			msg->sr_data = NULL;
		}
	}
	ebcd_pch_put(dpch);
	ebc_msg_free(msg);
	return err;
}

static ECM_INLINE ECMBOOL
need_bdcast_ack(ebc_pktb_t *pktb)
{
	ebc_ndid_t	ndid;

	ndid = pktb->pkthdr.mhdr.dstaddr.ndid;
	if (pktb->pchp != pchp_bdcast &&
	    pktb->pchp->ndid == ebcd_ndid &&
	    EBC_NDID_IS_BROADCAST(ndid) &&
	    (pktb->pkthdr.mhdr.msgt & EBC_MSGT_ONEWAY) == 0)
		return ECM_TRUE;
	return ECM_FALSE;
}

static ECM_INLINE ECMBOOL
is_my_pktb(ebc_pktb_t *pktb)
{
	ebc_addr_t	*daddr;
	ebc_ndid_t	ndid;

	daddr = &pktb->pkthdr.mhdr.dstaddr;
	ndid = daddr->ndid;
	if ((ndid == ebcd_ndid ||
	     (pktb->pchp == pchp_bdcast && EBC_NDID_IS_BROADCAST(ndid))) &&
	    daddr->mdid == EBC_MDID_EBCD)
		return ECM_TRUE;
	return ECM_FALSE;
}

void
ebcd_process_pktb(ebcd_pch_t *dpch, ebc_pktb_t *pktb)
{
	if (need_bdcast_ack(pktb)) {
		if (!ebcd_push_bdcastack(dpch, &pktb->pkthdr.mhdr)) {
			ebc_pktb_free(pktb);
			return;
		}
	}

	if (is_my_pktb(pktb)) {
		if (pktb->pkthdr.mhdr.msgt & EBC_MSGT_REQ)
			mch_put_inmsg_pktb(ebcd_mch, pktb);
		else
			mch_put_outmsg_pktb(ebcd_mch, pktb);
		return;
	}
	else
		ebcd_switch_pktb(dpch, pktb);
}

static void
start_loop_helpers(void)
{
	ebcd_start_detecter();
	ebcd_start_alivechecker();
	ebcd_start_router();
	ebcd_start_reader();
}

static void
stop_loop_helpers(void)
{
	ebcd_stop_reader();
	ebcd_stop_router();
	ebcd_stop_alivechecker();
	ebcd_stop_detecter();
}

void
ebcd_app_loop(void)
{
	ecm_uint32	timeout;

	start_loop_helpers();
	while (!going_to_shutdown) {
		timeout = EBCD_DEFAULT_TIMEOUT;
		PCH_SELECT(ebcd_selection, &timeout);

		ebcd_accept(ECM_FALSE);
		ebcd_read_pchs();
		ebcd_write_pchs();

		ebcd_wait_reader();
	}
	stop_loop_helpers();
}
